水位|下游_Flink中的时间和窗口

篇首语：本文由编程笔记#小编为大家整理，主要介绍了Flink 中的时间和窗口相关的知识，希望对你有一定的参考价值。

目录

6.1 时间语义

6.1.1 Flink 中的时间语义

6.1.2 哪种时间语义更重要

6.2 水位线&＃xff08;Watermark&＃xff09;

6.2.1 事件时间和窗口

6.2.2 什么是水位线

6.2.3 如何生成水位线

6.2.4 水位线的传递

 6.2.5 水位线的总结

6.1 时间语义

6.1.1 Flink 中的时间语义

Flink 是一个分布式处理 系统。分布式架构最大的特点&＃xff0c;就是节点彼此独立、互不影响&＃xff0c;这带来了更高的吞吐量和容错 性&＃xff1b;但有利必有弊&＃xff0c;最大的问题也来源于此。

在分布式系统中&＃xff0c;节点“各自为政”&＃xff0c;是没有统一时钟的&＃xff0c;数据和控制信息都通过网络进行传输。比如现在有一个任务是窗口聚合&＃xff0c;我们希望将每个小时的数据收集起来进行统计处理。 而对于并行的窗口子任务&＃xff0c;它们所在节点不同&＃xff0c;系统时间也会有差异&＃xff1b;当我们希望统计 8 点~9 点的数据时&＃xff0c;对并行任务来说其实并不是“同时”的&＃xff0c;收集到的数据也会有误差。

那既然一个集群中有 JobManager 作为管理者&＃xff0c;是不是让它统一向所有 TaskManager 发送 同步时钟信号就行了呢&＃xff1f;这也是不行的。因为网络传输会有延迟&＃xff0c;而且这延迟是不确定的&＃xff0c;所 以 JobManager 发出的同步信号无法同时到达所有节点&＃xff1b;想要拥有一个全局统一的时钟&＃xff0c;在分布式系统里是做不到的。

很明显&＃xff0c;这里有两个非常重要的时间点&＃xff1a;一个是数据产生的时间&＃xff0c;我们把它叫作“事件时间”&＃xff08;Event Time&＃xff09;&＃xff1b;另一个是数据真正被处理的时刻&＃xff0c;叫作“处理时间”&＃xff08;Processing Time&＃xff09;。 我们所定义的窗口操作&＃xff0c;到底是以那种时间作为衡量标准&＃xff0c;就是所谓的“时间语义”&＃xff08;Notions of Time&＃xff09;。由于分布式系统中网络传输的延迟和时钟漂移&＃xff0c;处理时间相对事件发生的时间会有所滞后。

 1. 处理时间&＃xff08;Processing Time&＃xff09;

处理时间的概念非常简单&＃xff0c;就是指执行处理操作的机器的系统时间。

如果我们以它作为衡量标准&＃xff0c;那么数据属于哪个窗口就很明显了&＃xff1a;只看窗口任务处理这条 数据时&＃xff0c;当前的系统时间。比如之前举的例子&＃xff0c;数据 8 点 59 分 59 秒产生&＃xff0c;而窗口计算时的时 间是 9 点零 1 秒&＃xff0c;那么这条数据就属于 9 点—10 点的窗口&＃xff1b;如果数据传输非常快&＃xff0c;9 点之前就 到了窗口任务&＃xff0c;那么它就属于 8 点—9 点的窗口了。每个并行的窗口子任务&＃xff0c;就只按照自己的 系统时钟划分窗口。假如我们在早上 8 点 10 分启动运行程序&＃xff0c;那么接下来一直到 9 点以前处 理的所有数据&＃xff0c;都属于第一个窗口&＃xff1b;9 点之后、10 点之前的所有数据就将属于第二个窗口。

这种方法非常简单粗暴&＃xff0c;不需要各个节点之间进行协调同步&＃xff0c;也不需要考虑数据在流中的 位置&＃xff0c;简单来说就是“我的地盘听我的”。所以处理时间是最简单的时间语义。

2. 事件时间&＃xff08;Event Time&＃xff09;

事件时间&＃xff0c;是指每个事件在对应的设备上发生的时间&＃xff0c;也就是数据生成的时间。

数据一旦产生&＃xff0c;这个时间自然就确定了&＃xff0c;所以它可以作为一个属性嵌入到数据中。这其实 就是这条数据记录的“时间戳”&＃xff08;Timestamp&＃xff09;。

在事件时间语义下&＃xff0c;我们对于时间的衡量&＃xff0c;就不看任何机器的系统时间了&＃xff0c;而是依赖于数据本身。打个比方&＃xff0c;这相当于任务处理的时候自己本身是没有时钟的&＃xff0c;所以只好来一个数据就 问一下“现在几点了”&＃xff1b;而数据本身也没有表&＃xff0c;只有一个自带的“出厂时间”&＃xff0c;于是任务就基于这 个时间来确定自己的时钟。由于流处理中数据是源源不断产生的&＃xff0c;一般来说&＃xff0c;先产生的数据也 会先被处理&＃xff0c;所以当任务不停地接到数据时&＃xff0c;它们的时间戳也基本上是不断增长的&＃xff0c;就可以代 表时间的推进。

当然我们会发现&＃xff0c;这里有个前提&＃xff0c;就是“先产生的数据先被处理”&＃xff0c;这要求我们可以保证 数据到达的顺序。但是由于分布式系统中网络传输延迟的不确定性&＃xff0c;实际应用中我们要面对的 数据流往往是乱序的。在这种情况下&＃xff0c;就不能简单地把数据自带的时间戳当作时钟了&＃xff0c;而需要 用另外的标志来表示事件时间进展&＃xff0c;在 Flink 中把它叫作事件时间的“水位线”&＃xff08;Watermarks&＃xff09;。 关于水位线的概念和用法&＃xff0c;我们会稍后介绍。

6.1.2 哪种时间语义更重要

1. 数据处理系统中的时间语义

在计算机系统中&＃xff0c;考虑数据处理的“时代变化”是没什么意义的&＃xff0c;我们更关心的&＃xff0c;显然是 数据本身产生的时间。

比如我们计算网站的 PV、UV 等指标&＃xff0c;要统计每天的访问量。如果某个用户在 23 点 59分 59 秒有一次访问&＃xff0c;但我们的任务处理这条数据的时间已经是第二天 0 点 0 分 01 秒了&＃xff1b;那么 这条数据&＃xff0c;是应该算作当天的访问&＃xff0c;还是第二天的访问呢&＃xff1f;很明显&＃xff0c;统计用户行为&＃xff0c;需要考虑 行为本身发生的时间&＃xff0c;所以我们应该把这条数据统计入当天的访问量。这时我们用到的窗口&＃xff0c; 就是以事件时间作为划分标准的&＃xff0c;跟处理时间无关。

所以在实际应用中&＃xff0c;事件时间语义会更为常见。一般情况下&＃xff0c;业务日志数据中都会记录数据生成的时间戳&＃xff08;timestamp&＃xff09;&＃xff0c;它就可以作为事件时间的判断基础。

2. 两种时间语义的对比

实际应用中&＃xff0c;数据产生的时间和处理的时间可能是完全不同的。很长时间收集起来的数据&＃xff0c; 处理或许只要一瞬间&＃xff1b;也有可能数据量过大、处理能力不足&＃xff0c;短时间堆了大量数据处理不完&＃xff0c; 产生“背压”&＃xff08;back pressure&＃xff09;。

通常来说&＃xff0c;处理时间是我们计算效率的衡量标准&＃xff0c;而事件时间会更符合我们的业务计算逻 辑。所以更多时候我们使用事件时间&＃xff1b;不过处理时间也不是一无是处。对于处理时间而言&＃xff0c;由 于没有任何附加考虑&＃xff0c;数据一来就直接处理&＃xff0c;因此这种方式可以让我们的流处理延迟降到最低&＃xff0c; 效率达到最高。

但是我们前面提到过&＃xff0c;在分布式环境中&＃xff0c;处理时间其实是不确定的&＃xff0c;各个并行任务时钟不 统一&＃xff1b;而且由于网络延迟&＃xff0c;导致数据到达各个算子任务的时间有快有慢&＃xff0c;对于窗口操作就可能 收集不到正确的数据了&＃xff0c;数据处理的顺序也会被打乱。这就会影响到计算结果的正确性。所以 处理时间语义&＃xff0c;一般用在对实时性要求极高、而对计算准确性要求不太高的场景。

而在事件时间语义下&＃xff0c;水位线成为了时钟&＃xff0c;可以统一控制时间的进度。这就保证了我们总可以将数据划分到正确的窗口中&＃xff0c;比如 8 点 59 分 59 秒产生的数据&＃xff0c;无论网络传输的延迟是多 少&＃xff0c;它永远属于 8 点~9 点的窗口&＃xff0c;不会错分。但我们知道数据还可能是乱序的&＃xff0c;要想让窗口 正确地收集到所有数据&＃xff0c;就必须等这些错乱的数据都到齐&＃xff0c;这就需要一定的等待时间。所以整 体上看&＃xff0c;事件时间语义是以一定延迟为代价&＃xff0c;换来了处理结果的正确性。由于网络延迟一般只 有毫秒级&＃xff0c;所以即使是事件时间语义&＃xff0c;同样可以完成低延迟实时流处理的任务。

另外&＃xff0c;除了事件时间和处理时间&＃xff0c;Flink 还有一个“摄入时间”&＃xff08;Ingestion Time&＃xff09;的概念&＃xff0c; 它是指数据进入 Flink 数据流的时间&＃xff0c;也就是 Source 算子读入数据的时间。摄入时间相当于是 事件时间和处理时间的一个中和&＃xff0c;它是把 Source 任务的处理时间&＃xff0c;当作了数据的产生时间添 加到数据里。这样一来&＃xff0c;水位线&＃xff08;watermark&＃xff09;也就基于这个时间直接生成&＃xff0c;不需要单独指定 了。这种时间语义可以保证比较好的正确性&＃xff0c;同时又不会引入太大的延迟。它的具体行为跟事 件时间非常像&＃xff0c;可以当作特殊的事件时间来处理。

在 Flink 中&＃xff0c;由于处理时间比较简单&＃xff0c;早期版本默认的时间语义是处理时间&＃xff1b;而考虑到事件时间在实际应用中更为广泛&＃xff0c;从 1.12 版本开始&＃xff0c;Flink 已经将事件时间作为了默认的时间语 义。

6.2 水位线&＃xff08;Watermark&＃xff09;

6.2.1 事件时间和窗口

在实际应用中&＃xff0c;一般会采用事件时间语义。而水位线&＃xff0c;就是基于事件时间提出的概念。所 以在介绍水位线之前&＃xff0c;我们首先来梳理一下事件时间和窗口的关系。

一个数据产生的时刻&＃xff0c;就是流处理中事件触发的时间点&＃xff0c;这就是“事件时间”&＃xff0c;一般都会以时间戳的形式作为一个字段记录在数据里。这个时间就像商品的“生产日期”一样&＃xff0c;一旦产 生就是固定的&＃xff0c;印在包装袋上&＃xff0c;不会因为运输辗转而变化。如果我们想要统计一段时间内的数 据&＃xff0c;需要划分时间窗口&＃xff0c;这时只要判断一下时间戳就可以知道数据属于哪个窗口了。

明确了一个数据的所属窗口&＃xff0c;还不能直接进行计算。因为窗口处理的是有界数据&＃xff0c;我们需要等窗口的数据都到齐了&＃xff0c;才能计算出最终的统计结果。那什么时候数据就都到齐了呢&＃xff1f;对于 时间窗口来说这很明显&＃xff1a;到了窗口的结束时间&＃xff0c;自然就应该收集到了所有数据&＃xff0c;就可以触发计 算输出结果了。比如我们想统计 8 点~9 点的用户点击量&＃xff0c;那就是从 8 点开始收集数据&＃xff0c;到 9点截止&＃xff0c;将收集的数据做处理计算。这有点类似于班车&＃xff0c;如图 6-3 所示&＃xff0c;每小时发一班&＃xff0c;那么8 点之后来的人都会上同一班车&＃xff0c;到 9 点钟准时发车&＃xff1b;9 点之后来的人&＃xff0c;就只好等下一班 10点发的车了。

在处理时间语义下&＃xff0c;都是以当前任务所在节点的系统时间为准的。这就相当于每辆车里都 挂了一个钟&＃xff0c;司机看到到了 9 点就直接发车。这种方式简单粗暴容易实现&＃xff0c;但因为车上的钟是 独立运行的&＃xff0c;以它为标准就不能准确地判断商品的生产时间。在分布式环境下&＃xff0c;这样会因为网 络传输延迟的不确定而导致误差。比如有些商品在 8 点 59 分 59 秒生产出来&＃xff0c;可是从下生产线 到运至车上又要花费几秒&＃xff0c;那就赶不上 9 点钟这班车了。而且现在分布式系统中有很多辆 9点发的班车&＃xff0c;所以同时生产出的一批商品&＃xff0c;需要平均分配到不同班车上&＃xff0c;可这些班车距离有近 有远、上面挂的钟有快有慢&＃xff0c;这就可能导致有些商品上车了、有些却被漏掉&＃xff1b;先后生产出的商 品&＃xff0c;到达车上的顺序也可能乱掉&＃xff1a;统计结果的正确性受到了影响。

所以在实际中我们往往需要以事件时间为准。如果考虑事件时间&＃xff0c;情况就复杂起来了。现 在不能直接用每辆车上挂的钟&＃xff08;系统时间&＃xff09;&＃xff0c;又没有统一的时钟&＃xff0c;那该怎么确定发车时间呢&＃xff1f;

现在能利用的&＃xff0c;就只有商品的生产时间&＃xff08;数据的时间戳&＃xff09;了。我们可以这样思考&＃xff1a;一般情 况下&＃xff0c;商品生产出来之后&＃xff0c;就会立即传送到车上&＃xff1b;所以商品到达车上的时间&＃xff08;系统时间&＃xff09;应该稍稍滞后于商品的生产时间&＃xff08;数据时间戳&＃xff09;。如果不考虑传输过程的一点点延迟&＃xff0c;我们就可以 直接用商品生产时间来表示当前车上的时间了。如图 6-4 所示&＃xff0c;到达车上的商品&＃xff0c;生产时间是8 点 05 分&＃xff0c;那么当前车上的时间就是 8 点 05 分&＃xff1b;又来了一个 8 点 10 分生产的商品&＃xff0c;现在车 上的时间就是 8 点 10 分。我们直接用数据的时间戳来指示当前的时间进展&＃xff0c;窗口的关闭自然 也是以数据的时间戳等于窗口结束时间为准&＃xff0c;这就相当于可以不受网络传输延迟的影响了。像 之前所说 8 点 59 分 59 秒生产出来的商品&＃xff0c;到车上的时候不管实际时间&＃xff08;系统时间&＃xff09;是几点&＃xff0c; 我们就认为当前是 8 点 59 分 59 秒&＃xff0c;所以它总是能赶上车的&＃xff1b;而 9 点这班车&＃xff0c;要等到 9 点整生 产的商品到来&＃xff0c;才认为时间到了 9 点&＃xff0c;这时才正式发车。这样就可以得到正确的统计结果了。

在这个处理过程中&＃xff0c;我们其实是基于数据的时间戳&＃xff0c;自定义了一个“逻辑时钟”。这个时钟的时间不会自动流逝&＃xff1b;它的时间进展&＃xff0c;就是靠着新到数据的时间戳来推动的。这样的好处在于&＃xff0c;计算的过程可以完全不依赖处理时间&＃xff08;系统时间&＃xff09;&＃xff0c;不论什么时候进行统计处理&＃xff0c;得到的结果都是正确的。比如双十一的时候系统处理压力大&＃xff0c;我们可能会把大量数据缓存在 Kafka中&＃xff1b;过了高峰时段之后再读取出来&＃xff0c;在几秒之内就可以处理完几个小时甚至几天的数据&＃xff0c;而且依然可以按照数据产生的时间段进行统计&＃xff0c;所有窗口都能收集到正确的数据。而一般实时流处 理的场景中&＃xff0c;事件时间可以基本与处理时间保持同步&＃xff0c;只是略微有一点延迟&＃xff0c;同时保证了窗口 计算的正确性。

6.2.2 什么是水位线

在事件时间语义下&＃xff0c;我们不依赖系统时间&＃xff0c;而是基于数据自带的时间戳去定义了一个时钟&＃xff0c; 用来表示当前时间的进展。于是每个并行子任务都会有一个自己的逻辑时钟&＃xff0c;它的前进是靠数 据的时间戳来驱动的。

但在分布式系统中&＃xff0c;这种驱动方式又会有一些问题。因为数据本身在处理转换的过程中会 变化&＃xff0c;如果遇到窗口聚合这样的操作&＃xff0c;其实是要攒一批数据才会输出一个结果&＃xff0c;那么下游的数据就会变少&＃xff0c;时间进度的控制就不够精细了。另外&＃xff0c;数据向下游任务传递时&＃xff0c;一般只能传输给一个子任务&＃xff08;除广播外&＃xff09;&＃xff0c;这样其他的并行子任务的时钟就无法推进了。例如一个时间戳为 9点整的数据到来&＃xff0c;当前任务的时钟就已经是 9 点了&＃xff1b;处理完当前数据要发送到下游&＃xff0c;如果下游任务是一个窗口计算&＃xff0c;并行度为 3&＃xff0c;那么接收到这个数据的子任务&＃xff0c;时钟也会进展到 9 点&＃xff0c;9点结束的窗口就可以关闭进行计算了&＃xff1b;而另外两个并行子任务则时间没有变化&＃xff0c;不能进行窗口计算。

所以我们应该把时钟也以数据的形式传递出去&＃xff0c;告诉下游任务当前时间的进展&＃xff1b;而且这个 时钟的传递不会因为窗口聚合之类的运算而停滞。一种简单的想法是&＃xff0c;在数据流中加入一个时钟标记&＃xff0c;记录当前的事件时间&＃xff1b;这个标记可以直接广播到下游&＃xff0c;当下游任务收到这个标记&＃xff0c;就 可以更新自己的时钟了。由于类似于水流中用来做标志的记号&＃xff0c;在 Flink 中&＃xff0c;这种用来衡量事 件时间&＃xff08;Event Time&＃xff09;进展的标记&＃xff0c;就被称作“水位线”&＃xff08;Watermark&＃xff09;。

具体实现上&＃xff0c;水位线可以看作一条特殊的数据记录&＃xff0c;它是插入到数据流中的一个标记点&＃xff0c; 主要内容就是一个时间戳&＃xff0c;用来指示当前的事件时间。而它插入流中的位置&＃xff0c;就应该是在某个数据到来之后&＃xff1b;这样就可以从这个数据中提取时间戳&＃xff0c;作为当前水位线的时间戳了。

如图 6-5 所示&＃xff0c;每个事件产生的数据&＃xff0c;都包含了一个时间戳&＃xff0c;我们直接用一个整数表示。 这里没有指定单位&＃xff0c;可以理解为秒或者毫秒&＃xff08;方便起见&＃xff0c;下面讲述统一认为是秒&＃xff09;。当产生于2 秒的数据到来之后&＃xff0c;当前的事件时间就是 2 秒&＃xff1b;在后面插入一个时间戳也为 2 秒的水位线&＃xff0c; 随着数据一起向下游流动。而当 5 秒产生的数据到来之后&＃xff0c;同样在后面插入一个水位线&＃xff0c;时间 戳也为 5&＃xff0c;当前的时钟就推进到了 5 秒。这样&＃xff0c;如果出现下游有多个并行子任务的情形&＃xff0c;我们 只要将水位线广播出去&＃xff0c;就可以通知到所有下游任务当前的时间进度了。

水位线就像它的名字所表达的&＃xff0c;是数据流中的一部分&＃xff0c;随着数据一起流动&＃xff0c;在不同任务之间传输。这看起来非常简单&＃xff1b;接下来我们就进一步探讨一些复杂的状况。

1. 有序流中的水位线

在理想状态下&＃xff0c;数据应该按照它们生成的先后顺序、排好队进入流中&＃xff1b;也就是说&＃xff0c;它们处理的过程会保持原先的顺序不变&＃xff0c;遵守先来后到的原则。这样的话我们从每个数据中提取时间 戳&＃xff0c;就可以保证总是从小到大增长的&＃xff0c;从而插入的水位线也会不断增长、事件时钟不断向前推 进。

实际应用中&＃xff0c;如果当前数据量非常大&＃xff0c;可能会有很多数据的时间戳是相同的&＃xff0c;这时每来一 条数据就提取时间戳、插入水位线就做了大量的无用功。而且即使时间戳不同&＃xff0c;同时涌来的数 据时间差会非常小&＃xff08;比如几毫秒&＃xff09;&＃xff0c;往往对处理计算也没什么影响。所以为了提高效率&＃xff0c;一般 会每隔一段时间生成一个水位线&＃xff0c;这个水位线的时间戳&＃xff0c;就是当前最新数据的时间戳&＃xff0c;如图6-6 所示。所以这时的水位线&＃xff0c;其实就是有序流中的一个周期性出现的时间标记。

 这里需要注意的是&＃xff0c;水位线插入的“周期”&＃xff0c;本身也是一个时间概念。在当前事件时间语 义下&＃xff0c;假如我们设定了每隔 100ms 生成一次水位线&＃xff0c;那就是要等事件时钟推进 100ms 才能插入&＃xff1b;但是事件时钟本身的进展&＃xff0c;本身就是靠水位线来表示的——现在要插入一个水位线&＃xff0c;可前 提又是水位线要向前推进 100ms&＃xff0c;这就陷入了死循环。所以对于水位线的周期性生成&＃xff0c;周期时间是指处理时间&＃xff08;系统时间&＃xff09;&＃xff0c;而不是事件时间。

2. 乱序流中的水位线

有序流的处理非常简单&＃xff0c;看起来水位线也并没有起到太大的作用。但这种情况只存在于理 想状态下。我们知道在分布式系统中&＃xff0c;数据在节点间传输&＃xff0c;会因为网络传输延迟的不确定性&＃xff0c; 导致顺序发生改变&＃xff0c;这就是所谓的“乱序数据”。

这里所说的“乱序”&＃xff08;out-of-order&＃xff09;&＃xff0c;是指数据的先后顺序不一致&＃xff0c;主要就是基于数据的产 生时间而言的。如图 6-7 所示&＃xff0c;一个 7 秒时产生的数据&＃xff0c;生成时间自然要比 9 秒的数据早&＃xff1b;但 是经过数据缓存和传输之后&＃xff0c;处理任务可能先收到了 9 秒的数据&＃xff0c;之后 7 秒的数据才姗姗来迟。 这时如果我们希望插入水位线&＃xff0c;来指示当前的事件时间进展&＃xff0c;又该怎么做呢&＃xff1f;

最直观的想法自然是跟之前一样&＃xff0c;我们还是靠数据来驱动&＃xff0c;每来一个数据就提取它的时间 戳、插入一个水位线。不过现在的情况是数据乱序&＃xff0c;所以有可能新的时间戳比之前的还小&＃xff0c;如 果直接将这个时间的水位线再插入&＃xff0c;我们的“时钟”就回退了——水位线就代表了时钟&＃xff0c;时光 不能倒流&＃xff0c;所以水位线的时间戳也不能减小。

解决思路也很简单&＃xff1a;

我们插入新的水位线时&＃xff0c;要先判断一下时间戳是否比之前的大&＃xff0c;否则 就不再生成新的水位线&＃xff0c;如图 6-8 所示。也就是说&＃xff0c;只有数据的时间戳比当前时钟大&＃xff0c;才能推动时钟前进&＃xff0c;这时才插入水位线。

如果考虑到大量数据同时到来的处理效率&＃xff0c;我们同样可以周期性地生成水位线。这时只需要保存一下之前所有数据中的最大时间戳&＃xff0c;需要插入水位线时&＃xff0c;就直接以它作为时间戳生成新的水位线

这样做尽管可以定义出一个事件时钟&＃xff0c;却也会带来一个非常大的问题&＃xff1a;我们无法正确处理 “迟到”的数据。在上面的例子中&＃xff0c;当 9 秒产生的数据到来之后&＃xff0c;我们就直接将时钟推进到了9 秒&＃xff1b;如果有一个窗口结束时间就是 9 秒&＃xff08;比如&＃xff0c;要统计 0~9 秒的所有数据&＃xff09;&＃xff0c;那么这时窗口就应该关闭、将收集到的所有数据计算输出结果了。但事实上&＃xff0c;由于数据是乱序的&＃xff0c;还可能有 时间戳为 7 秒、8 秒的数据在 9 秒的数据之后才到来&＃xff0c;这就是“迟到数据”&＃xff08;late data&＃xff09;。它们 本来也应该属于 0~9 秒这个窗口&＃xff0c;但此时窗口已经关闭&＃xff0c;于是这些数据就被遗漏了&＃xff0c;这会导致统计结果不正确。

为了让窗口能够正确收集到迟到的数据&＃xff0c;我们也可以等上 2 秒&＃xff1b;也就是 用当前已有数据的最大时间戳减去 2 秒&＃xff0c;就是要插入的水位线的时间戳&＃xff0c;如图 6-10 所示。这 样的话&＃xff0c;9 秒的数据到来之后&＃xff0c;事件时钟不会直接推进到 9 秒&＃xff0c;而是进展到了 7 秒&＃xff1b;必须等到11 秒的数据到来之后&＃xff0c;事件时钟才会进展到 9 秒&＃xff0c;这时迟到数据也都已收集齐&＃xff0c;0~9 秒的窗 口就可以正确计算结果了。

为了防止“迟到数据”还没有传输过来&＃xff0c;我们可以选择再多等几秒&＃xff0c;以计算出正确的答案。

 3. 水位线的特性

我们可以总结一下水位线的特性&＃xff1a;

⚫ 水位线是插入到数据流中的一个标记&＃xff0c;可以认为是一个特殊的数据

⚫ 水位线主要的内容是一个时间戳&＃xff0c;用来表示当前事件时间的进展

⚫ 水位线是基于数据的时间戳生成的

⚫ 水位线的时间戳必须单调递增&＃xff0c;以确保任务的事件时间时钟一直向前推进

⚫ 水位线可以通过设置延迟&＃xff0c;来保证正确处理乱序数据

⚫ 一个水位线 Watermark(t)&＃xff0c;表示在当前流中事件时间已经达到了时间戳 t, 这代表 t 之 前的所有数据都到齐了&＃xff0c;之后流中不会出现时间戳 t’ ≤ t 的数据

水位线是 Flink 流处理中保证结果正确性的核心机制&＃xff0c;它往往会跟窗口一起配合&＃xff0c;完成对 乱序数据的正确处理。

6.2.3 如何生成水位线

1. 生成水位线的总体原则

我们知道&＃xff0c;完美的水位线是“绝对正确”的&＃xff0c;也就是一个水位线一旦出现&＃xff0c;就表示这个时间之前的数据已经全部到齐、之后再也不会出现了。而完美的东西总是可望不可即&＃xff0c;我们只能 尽量去保证水位线的正确。如果对结果正确性要求很高、想要让窗口收集到所有数据&＃xff0c;我们该 怎么做呢&＃xff1f;

可以单独创建一个 Flink 作业来监控事件流&＃xff0c;建立概率分布或者机器学习模型&＃xff0c;学习事件的迟到规律。得到分布规律之后&＃xff0c;就可以选择置信区间来确定延迟&＃xff0c;作为水位线的生成策略了。例如&＃xff0c;如果得到数据的迟到时间服从μ&＃61;1&＃xff0c;σ&＃61;1 的正态分布&＃xff0c;那么设置水位线延迟为 3 秒&＃xff0c;就可以保证至少 97.7%的数据可以正确处理。

如果我们希望计算结果能更加准确&＃xff0c;那可以将水位线的延迟设置得更高一些&＃xff0c;等待的时间越长&＃xff0c;自然也就越不容易漏掉数据。不过这样做的代价是处理的实时性降低了&＃xff0c;我们可能为极少数的迟到数据增加了很多不必要的延迟。

如果我们希望处理得更快、实时性更强&＃xff0c;那么可以将水位线延迟设得低一些。这种情况下&＃xff0c; 可能很多迟到数据会在水位线之后才到达&＃xff0c;就会导致窗口遗漏数据&＃xff0c;计算结果不准确。对于这 些 “漏网之鱼”&＃xff0c;Flink 另外提供了窗口处理迟到数据的方法&＃xff0c;我们会在后面介绍。当然&＃xff0c;如果我们对准确性完全不考虑、一味地追求处理速度&＃xff0c;可以直接使用处理时间语义&＃xff0c;这在理论上可以得到最低的延迟。

2. 水位线生成策略&＃xff08;Watermark Strategies&＃xff09;

在 Flink 的 DataStream API 中 &＃xff0c; 有 一 个 单 独 用 于 生 成 水 位 线 的 方 法&＃xff1a;.assignTimestampsAndWatermarks()&＃xff0c;它主要用来为流中的数据分配时间戳&＃xff0c;并生成水位线 来指示事件时间&＃xff1a;

public SingleOutputStreamOperator assignTimestampsAndWatermarks(
WatermarkStrategy watermarkStrategy)

具体使用时&＃xff0c;直接用 DataStream 调用该方法即可&＃xff0c;与普通的 transform 方法完全一样。

DataStream stream &＃61; env.addSource(new ClickSource());
DataStream withTimestampsAndWatermarks &＃61;
stream.assignTimestampsAndWatermarks();

这里需要将原始的时间戳提取出来&＃xff0c;如果不提取出来并明确把它分配给数据&＃xff0c;Flink是无法知道数据真正产生的时间的。当然&＃xff0c;有些时候数据源本身就提供了时间戳信息&＃xff0c; 比如读取 Kafka 时&＃xff0c;我们就可以从 Kafka 数据中直接获取时间戳&＃xff0c;而不需要单独提取字段分配 了。

.assignTimestampsAndWatermarks()方法需要传入一个 WatermarkStrategy 作为参数&＃xff0c;这就 是 所 谓 的 “ 水 位 线 生 成 策 略 ” 。 WatermarkStrategy 中 包 含 了 一 个 “ 时间戳分配器”TimestampAssigner 和一个“水位线生成器”WatermarkGenerator。

⚫ TimestampAssigner&＃xff1a;主要负责从流中数据元素的某个字段中提取时间戳&＃xff0c;并分配给 元素。时间戳的分配是生成水位线的基础。

⚫ WatermarkGenerator&＃xff1a;主要负责按照既定的方式&＃xff0c;基于时间戳生成水位线。在

WatermarkGenerator 接口中&＃xff0c;主要又有两个方法&＃xff1a;onEvent()和 onPeriodicEmit()。

⚫ onEvent&＃xff1a;每个事件&＃xff08;数据&＃xff09;到来都会调用的方法&＃xff0c;它的参数有当前事件、时间戳&＃xff0c; 以及允许发出水位线的一个 WatermarkOutput&＃xff0c;可以基于事件做各种操作

⚫ onPeriodicEmit&＃xff1a;周期性调用的方法&＃xff0c;可以由 WatermarkOutput 发出水位线。周期时间 为处理时间&＃xff0c;可以调用环境配置的.setAutoWatermarkInterval()方法来设置&＃xff0c;默认为200ms。

env.getConfig().setAutoWatermarkInterval(60 * 1000L);


3. Flink 内置水位线生成器

WatermarkStrategy 这个接口是一个生成水位线策略的抽象&＃xff0c;让我们可以灵活地实现自己的需求&＃xff1b;但看起来有些复杂&＃xff0c;如果想要自己实现应该还是比较麻烦的。好在 Flink 充分考虑到了 我们的痛苦&＃xff0c;提供了内置的水位线生成器&＃xff08;WatermarkGenerator&＃xff09;&＃xff0c;不仅开箱即用简化了编程&＃xff0c; 而且也为我们自定义水位线策略提供了模板。

这两个生成器可以通过调用 WatermarkStrategy 的静态辅助方法来创建。它们都是周期性生成水位线的&＃xff0c;分别对应着处理有序流和乱序流的场景。

&＃xff08;1&＃xff09;有序流

对于有序流&＃xff0c;主要特点就是时间戳单调增长&＃xff08;Monotonously Increasing Timestamps&＃xff09;&＃xff0c;所以 永远不会出现迟到数据的问题。这是周期性生成水位线的最简单的场景&＃xff0c;直接调用WatermarkStrategy.forMonotonousTimestamps()方法就可以实现。简单来说&＃xff0c;就是直接拿当前最大的时间戳作为水位线就可以了。


有序&＃xff1a;
调用.withTimestampAssigner()方法&＃xff0c;将数据中的 timestamp 字段提取出来&＃xff0c;
作为时间戳分配给数据元素&＃xff1b;然后用内置的有序流水位线生成器构造出了生成策略。
这样&＃xff0c;提取出的数据时间戳&＃xff0c;就是我们处理计算的事件时间。
这里需要注意的是&＃xff0c;时间戳和水位线的单位&＃xff0c;必须都是毫秒。

.assignTimestampsAndWatermarks(WatermarkStrategy.forMonotonousTimestamps()
.withTimestampAssigner(new SerializableTimestampAssigner()
&＃64;Override
public long extractTimestamp(Event event, long l)
return event.timestamp;
)
)


&＃xff08;2&＃xff09;乱序流

由于乱序流中需要等待迟到数据到齐&＃xff0c;所以必须设置一个固定量的延迟时间&＃xff08;Fixed Amount of Lateness&＃xff09;。这时生成水位线的时间戳&＃xff0c;就是当前数据流中最大的时间戳减去延迟的结果&＃xff0c;相当于把表调慢&＃xff0c;当前时钟会滞后于数据的最大时间戳。调用 WatermarkStrategy. forBoundedOutOfOrderness()方法就可以实现。这个方法需要传入一个 maxOutOfOrderness 参 数&＃xff0c;表示“最大乱序程度”&＃xff0c;它表示数据流中乱序数据时间戳的最大差值&＃xff1b;如果我们能确定乱序 程度&＃xff0c;那么设置对应时间长度的延迟&＃xff0c;就可以等到所有的乱序数据了。

//乱序流的WaterMark生成
//1.WaterMark的生成器
.assignTimestampsAndWatermarks(WatermarkStrategy.forBoundedOutOfOrderness(Duration.ofSeconds(2))
.withTimestampAssigner(new SerializableTimestampAssigner()
&＃64;Override
//2.时间戳的提取器
public long extractTimestamp(Event event, long l)
return event.timestamp;
)
);

上面代码中&＃xff0c;我们同样提取了 timestamp 字段作为时间戳&＃xff0c;并且以 5 秒的延迟时间创建了 处理乱序流的水位线生成器。事实上&＃xff0c;有序流的水位线生成器本质上和乱序流是一样的&＃xff0c;相当于延迟设为 0 的乱序流水 位线生成器&＃xff0c;两者完全等同&＃xff1a;

WatermarkStrategy.forMonotonousTimestamps()
WatermarkStrategy.forBoundedOutOfOrderness(Duration.ofSeconds(0))


4. 自定义水位线策略

一般来说&＃xff0c;Flink 内置的水位线生成器就可以满足应用需求了。不过有时我们的业务逻辑 可能非常复杂&＃xff0c;这时对水位线生成的逻辑也有更高的要求&＃xff0c;我们就必须自定义实现水位线策略WatermarkStrategy 了。

在 WatermarkStrategy 中&＃xff0c;时间戳分配器 TimestampAssigner 都是大同小异的&＃xff0c;指定字段提 取时间戳就可以了&＃xff1b;而不同策略的关键就在于 WatermarkGenerator 的实现。整体说来&＃xff0c;Flink有两种不同的生成水位线的方式&＃xff1a;一种是周期性的&＃xff08;Periodic&＃xff09;&＃xff0c;另一种是断点式的&＃xff08;Punctuated&＃xff09;。

&＃xff08;1&＃xff09;周期性水位线生成器&＃xff08;Periodic Generator&＃xff09;

周期性生成器一般是通过 onEvent()观察判断输入的事件&＃xff0c;而在 onPeriodicEmit()里发出水 位线。

下面是一段自定义周期性生成水位线的代码&＃xff1a;

import org.apache.flink.api.common.eventtime.*;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

// 自定义水位线的产生
public class CustomWatermarkTest
public static void main(String[] args) throws Exception
StreamExecutionEnvironment env &＃61;
StreamExecutionEnvironment.getExecutionEnvironment();
env.setParallelism(1);

env
.addSource(new ClickSource())
.assignTimestampsAndWatermarks(new CustomWatermarkStrategy())
.print();

env.execute();


public static class CustomWatermarkStrategy implements WatermarkStrategy
&＃64;Override
public TimestampAssigner
createTimestampAssigner(TimestampAssignerSupplier.Context context)
return new SerializableTimestampAssigner()
&＃64;Override
136

public long extractTimestamp(Event element, long recordTimestamp)

return element.timestamp; // 告诉程序数据源里的时间戳是哪一个字段

;


&＃64;Override
public WatermarkGenerator
createWatermarkGenerator(WatermarkGeneratorSupplier.Context context)
return new CustomPeriodicGenerator();



public static class CustomPeriodicGenerator implements
WatermarkGenerator
private Long delayTime &＃61; 5000L; // 延迟时间
private Long maxTs &＃61; Long.MIN_VALUE &＃43; delayTime &＃43; 1L; // 观察到的最大时间戳

&＃64;Override
public void onEvent(Event event, long eventTimestamp, WatermarkOutput
output)
// 每来一条数据就调用一次
maxTs &＃61; Math.max(event.timestamp, maxTs); // 更新最大时间戳


&＃64;Override
public void onPeriodicEmit(WatermarkOutput output)
// 发射水位线&＃xff0c;默认 200ms 调用一次
output.emitWatermark(new Watermark(maxTs - delayTime - 1L));





我们在 onPeriodicEmit()里调用 output.emitWatermark()&＃xff0c;就可以发出水位线了&＃xff1b;这个方法 由系统框架周期性地调用&＃xff0c;默认 200ms 一次。所以水位线的时间戳是依赖当前已有数据的最 大时间戳的&＃xff08;这里的实现与内置生成器类似&＃xff0c;也是减去延迟时间再减 1&＃xff09;&＃xff0c;但具体什么时候生 成与数据无关。

&＃xff08;2&＃xff09;断点式水位线生成器&＃xff08;Punctuated Generator&＃xff09;

断点式生成器会不停地检测 onEvent()中的事件&＃xff0c;当发现带有水位线信息的特殊事件时&＃xff0c; 就立即发出水位线。一般来说&＃xff0c;断点式生成器不会通过 onPeriodicEmit()发出水位线。

自定义的断点式水位线生成器代码如下&＃xff1a;

public class CustomPunctuatedGenerator implements WatermarkGenerator
&＃64;Override
public void onEvent(Event r, long eventTimestamp, WatermarkOutput output)
// 只有在遇到特定的 itemId 时&＃xff0c;才发出水位线
if (r.user.equals("Mary"))
output.emitWatermark(new Watermark(r.timestamp - 1));



&＃64;Override
public void onPeriodicEmit(WatermarkOutput output)
// 不需要做任何事情&＃xff0c;因为我们在 onEvent 方法中发射了水位线




我们在 onEvent()中判断当前事件的 user 字段&＃xff0c;只有遇到“Mary”这个特殊的值时&＃xff0c;才调用output.emitWatermark()发出水位线。这个过程是完全依靠事件来触发的&＃xff0c;所以水位线的生成一 定在某个数据到来之后。

5. 在自定义数据源中发送水位线

我们也可以在自定义的数据源中抽取事件时间&＃xff0c;然后发送水位线。这里要注意的是&＃xff0c;在自 定义数据源中发送了水位线以后&＃xff0c;就不能再在程序中使用 assignTimestampsAndWatermarks 方 法 来 生 成 水 位 线 了 。 在 自 定 义 数 据 源 中 生 成 水 位 线 和 在 程 序 中 使 用

assignTimestampsAndWatermarks 方法生成水位线二者只能取其一。

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.source.SourceFunction;
import org.apache.flink.streaming.api.watermark.Watermark;


import java.util.Calendar;
import java.util.Random;

public class EmitWatermarkInSourceFunction
public static void main(String[] args) throws Exception
StreamExecutionEnvironment env &＃61;
StreamExecutionEnvironment.getExecutionEnvironment();
env.setParallelism(1);

env.addSource(new ClickSourceWithWatermark()).print();

env.execute();


// 泛型是数据源中的类型
public static class ClickSourceWithWatermark implements SourceFunction

private boolean running &＃61; true;
&＃64;Override
public void run(SourceContext sourceContext) throws Exception
Random random &＃61; new Random();
String[] userArr &＃61; "Mary", "Bob", "Alice";
String[] urlArr &＃61; "./home", "./cart", "./prod?id&＃61;1";
while (running)
long currTs &＃61; Calendar.getInstance().getTimeInMillis(); // 毫秒时
间戳
String username &＃61; userArr[random.nextInt(userArr.length)];
String url &＃61; urlArr[random.nextInt(urlArr.length)];
Event event &＃61; new Event(username, url, currTs);
// 使用 collectWithTimestamp 方法将数据发送出去&＃xff0c;并指明数据中的时间戳的字段
sourceContext.collectWithTimestamp(event, event.timestamp);
// 发送水位线
sourceContext.emitWatermark(new Watermark(event.timestamp - 1L));
Thread.sleep(1000L);



&＃64;Override
public void cancel()
running &＃61; false;





6.2.4 水位线的传递

&＃xff08;1&＃xff09;上游并行子任务发来不同的水位线&＃xff0c;当前任务会为每一个分区设置一个“分区水位线”

&＃xff08;Partition Watermark&＃xff09;&＃xff0c;这是一个分区时钟&＃xff1b;而当前任务自己的时钟&＃xff0c;就是所有分区时钟里最 小的那个。

&＃xff08;2&＃xff09;当有一个新的水位线&＃xff08;第一分区的 4&＃xff09;从上游传来时&＃xff0c;当前任务会首先更新对应的分 区时钟&＃xff1b;然后再次判断所有分区时钟中的最小值&＃xff0c;如果比之前大&＃xff0c;说明事件时间有了进展&＃xff0c;当 前任务的时钟也就可以更新了。这里要注意&＃xff0c;更新后的任务时钟&＃xff0c;并不一定是新来的那个分区 水位线&＃xff0c;比如这里改变的是第一分区的时钟&＃xff0c;但最小的分区时钟是第三分区的 3&＃xff0c;于是当前任 务时钟就推进到了 3。当时钟有进展时&＃xff0c;当前任务就会将自己的时钟以水位线的形式&＃xff0c;广播给 下游所有子任务。

&＃xff08;3&＃xff09;再次收到新的水位线&＃xff08;第二分区的 7&＃xff09;后&＃xff0c;执行同样的处理流程。首先将第二个分区 时钟更新为 7&＃xff0c;然后比较所有分区时钟&＃xff1b;发现最小值没有变化&＃xff0c;那么当前任务的时钟也不变&＃xff0c;140也不会向下游任务发出水位线。

&＃xff08;4&＃xff09;同样道理&＃xff0c;当又一次收到新的水位线&＃xff08;第三分区的 6&＃xff09;之后&＃xff0c;第三个分区时钟更新为6&＃xff0c;同时所有分区时钟最小值变成了第一分区的 4&＃xff0c;所以当前任务的时钟推进到 4&＃xff0c;并发出时间 戳为 4 的水位线&＃xff0c;广播到下游各个分区任务。

水位线在上下游任务之间的传递&＃xff0c;非常巧妙地避免了分布式系统中没有统一时钟的问题&＃xff0c; 每个任务都以“处理完之前所有数据”为标准来确定自己的时钟&＃xff0c;就可以保证窗口处理的结果 总是正确的。对于有多条流合并之后进行处理的场景&＃xff0c;水位线传递的规则是类似的。

 6.2.5 水位线的总结

水位线在事件时间的世界里面&＃xff0c;承担了时钟的角色。也就是说在事件时间的流中&＃xff0c;水位线 是唯一的时间尺度。如果想要知道现在几点&＃xff0c;就要看水位线的大小。后面讲到的窗口的闭合&＃xff0c; 以及定时器的触发都要通过判断水位线的大小来决定是否触发。

水位线是一种特殊的事件&＃xff0c;由程序员通过编程插入的数据流里面&＃xff0c;然后跟随数据流向下游 流动。

水位线的默认计算公式&＃xff1a;水位线 &＃61; 观察到的最大事件时间 – 最大延迟时间 – 1 毫秒。

在数据流开始之前&＃xff0c;Flink 会插入一个大小是负无穷大&＃xff08;在 Java 中是-Long.MAX_VALUE&＃xff09; 的水位线&＃xff0c;而在数据流结束时&＃xff0c;Flink 会插入一个正无穷大(Long.MAX_VALUE)的水位线&＃xff0c;保 证所有的窗口闭合以及所有的定时器都被触发。

对于离线数据集&＃xff0c;Flink 也会将其作为流读入&＃xff0c;也就是一条数据一条数据的读取。在这种 情况下&＃xff0c;Flink 对于离线数据集&＃xff0c;只会插入两次水位线&＃xff0c;也就是在最开始处插入负无穷大的水位线&＃xff0c;在结束位置插入一个正无穷大的水位线。因为只需要插入两次水位线&＃xff0c;就可以保证计算 的正确&＃xff0c;无需在数据流的中间插入水位线了。